{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# | hide\n",
    "\n",
    "from IPython.display import Markdown, display_markdown"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# | hide\n",
    "\n",
    "try:\n",
    "    import google.colab\n",
    "    in_colab = True\n",
    "except:\n",
    "    in_colab = False"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/markdown": [
       "\n",
       "### If you see this message, you are running in Google Colab\n",
       "Along with this interactive tutorial the content of this notebook is organized and formatted for documentation purpuoses. \n",
       "\n",
       "You can ignore the '# | hide', '# | notest' and '# | echo: false' comments, they are not important for the tutorial.\n",
       "    "
      ],
      "text/plain": [
       "<IPython.core.display.Markdown object>"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "# | hide\n",
    "\n",
    "if in_colab:\n",
    "    display(Markdown(\"\"\"\n",
    "### If you see this message, you are running in Google Colab\n",
    "Along with this interactive tutorial the content of this notebook is organized and formatted for documentation purpuoses. \n",
    "\n",
    "You can ignore the '# | hide', '# | notest' and '# | echo: false' comments, they are not important for the tutorial.\n",
    "    \"\"\"))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# | notest\n",
    "# | hide\n",
    "\n",
    "# This step is necessary to be able to run FastKafka inside Google Colab\n",
    "\n",
    "import nest_asyncio"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# | notest\n",
    "# | hide\n",
    "\n",
    "nest_asyncio.apply()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# FastKafka\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<b>Effortless Kafka integration for your web services</b>\n",
    "\n",
    "## Deprecation notice\n",
    "\n",
    "This project is superceeded by [FastStream](https://github.com/airtai/faststream).\n",
    "\n",
    "FastStream is a new package based on the ideas and experiences gained from\n",
    "[FastKafka](https://github.com/airtai/fastkafka) and\n",
    "[Propan](https://github.com/lancetnik/propan). By joining our forces, we\n",
    " picked up the best from both packages and created the unified way to write\n",
    "  services capable of processing streamed data regradless of the underliying protocol.\n",
    "\n",
    "  We'll continue to maintain FastKafka package, but new development will be in  [FastStream](https://github.com/airtai/faststream).\n",
    "If you are starting a new service,  [FastStream](https://github.com/airtai/faststream) is the recommended way to do it."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "\n",
    "---\n",
    "\n",
    "![PyPI](https://img.shields.io/pypi/v/fastkafka)\n",
    "![PyPI - Downloads](https://img.shields.io/pypi/dm/fastkafka)\n",
    "![PyPI - Python Version](https://img.shields.io/pypi/pyversions/fastkafka)\n",
    "\n",
    "![GitHub Workflow Status](https://img.shields.io/github/actions/workflow/status/airtai/fastkafka/test.yaml)\n",
    "![CodeQL](https://github.com/airtai/fastkafka//actions/workflows/codeql.yml/badge.svg)\n",
    "![Dependency Review](https://github.com/airtai/fastkafka//actions/workflows/dependency-review.yml/badge.svg)\n",
    "\n",
    "![GitHub](https://img.shields.io/github/license/airtai/fastkafka)\n",
    "\n",
    "---\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "[FastKafka](https://fastkafka.airt.ai/) is a powerful and easy-to-use Python library for building asynchronous services that interact with Kafka topics. Built on top of [Pydantic](https://docs.pydantic.dev/), [AIOKafka](https://github.com/aio-libs/aiokafka) and [AsyncAPI](https://www.asyncapi.com/), FastKafka simplifies the process of writing producers and consumers for Kafka topics, handling all the parsing, networking, task scheduling and data generation automatically. With FastKafka, you can quickly prototype and develop high-performance Kafka-based services with minimal code, making it an ideal choice for developers looking to streamline their workflow and accelerate their projects.\n",
    "\n",
    "---\n",
    "\n",
    "#### ⭐⭐⭐ Stay in touch ⭐⭐⭐\n",
    "\n",
    "Please show your support and stay in touch by:\n",
    "\n",
    "- giving our [GitHub repository](https://github.com/airtai/fastkafka/) a star, and\n",
    "\n",
    "- joining our [Discord server](https://discord.gg/CJWmYpyFbc).\n",
    "\n",
    "Your support helps us to stay in touch with you and encourages us to continue developing and improving the library. Thank you for your support!\n",
    "\n",
    "---\n",
    "\n",
    "#### 🐝🐝🐝 We were busy lately 🐝🐝🐝\n",
    "\n",
    "![Activity](https://repobeats.axiom.co/api/embed/21f36049093d5eb8e5fdad18c3c5d8df5428ca30.svg \"Repobeats analytics image\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Install\n",
    "\n",
    "FastKafka works on Windows, macOS, Linux, and most Unix-style operating systems. You can install base version of FastKafka with `pip` as usual:\n",
    "\n",
    "```sh\n",
    "pip install fastkafka\n",
    "```\n",
    "\n",
    "To install FastKafka with testing features please use:\n",
    "```sh\n",
    "pip install fastkafka[test]\n",
    "```\n",
    "\n",
    "To install FastKafka with asyncapi docs please use:\n",
    "```sh\n",
    "pip install fastkafka[docs]\n",
    "```\n",
    "\n",
    "To install FastKafka with all the features please use:\n",
    "```sh\n",
    "pip install fastkafka[test,docs]\n",
    "```"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Looking in indexes: https://pypi.org/simple, https://us-python.pkg.dev/colab-wheels/public/simple/\n",
      "Collecting fastkafka[docs,test]\n",
      "  Downloading fastkafka-0.6.1-py3-none-any.whl (91 kB)\n",
      "\u001b[2K     \u001b[90m━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━\u001b[0m \u001b[32m91.1/91.1 kB\u001b[0m \u001b[31m5.5 MB/s\u001b[0m eta \u001b[36m0:00:00\u001b[0m\n",
      "\u001b[?25hCollecting aiokafka>=0.8.0 (from fastkafka[docs,test])\n",
      "  Downloading aiokafka-0.8.0-cp310-cp310-manylinux_2_17_x86_64.manylinux2014_x86_64.whl (1.1 MB)\n",
      "\u001b[2K     \u001b[90m━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━\u001b[0m \u001b[32m1.1/1.1 MB\u001b[0m \u001b[31m33.6 MB/s\u001b[0m eta \u001b[36m0:00:00\u001b[0m\n",
      "\u001b[?25hRequirement already satisfied: anyio>=3.0 in /usr/local/lib/python3.10/dist-packages (from fastkafka[docs,test]) (3.6.2)\n",
      "Collecting asyncer>=0.0.2 (from fastkafka[docs,test])\n",
      "  Downloading asyncer-0.0.2-py3-none-any.whl (8.3 kB)\n",
      "Collecting docstring-parser>=0.15 (from fastkafka[docs,test])\n",
      "  Downloading docstring_parser-0.15-py3-none-any.whl (36 kB)\n",
      "Requirement already satisfied: nest-asyncio>=1.5.6 in /usr/local/lib/python3.10/dist-packages (from fastkafka[docs,test]) (1.5.6)\n",
      "Requirement already satisfied: pydantic>=1.9 in /usr/local/lib/python3.10/dist-packages (from fastkafka[docs,test]) (1.10.7)\n",
      "Requirement already satisfied: tqdm>=4.62 in /usr/local/lib/python3.10/dist-packages (from fastkafka[docs,test]) (4.65.0)\n",
      "Requirement already satisfied: typer>=0.7.0 in /usr/local/lib/python3.10/dist-packages (from fastkafka[docs,test]) (0.7.0)\n",
      "Collecting install-jdk==0.3.0 (from fastkafka[docs,test])\n",
      "  Downloading install-jdk-0.3.0.tar.gz (3.8 kB)\n",
      "  Preparing metadata (setup.py) ... \u001b[?25l\u001b[?25hdone\n",
      "Collecting ipywidgets<=8.0.4,>=8.0 (from fastkafka[docs,test])\n",
      "  Downloading ipywidgets-8.0.4-py3-none-any.whl (137 kB)\n",
      "\u001b[2K     \u001b[90m━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━\u001b[0m \u001b[32m137.8/137.8 kB\u001b[0m \u001b[31m14.1 MB/s\u001b[0m eta \u001b[36m0:00:00\u001b[0m\n",
      "\u001b[?25hRequirement already satisfied: requests>=2.20 in /usr/local/lib/python3.10/dist-packages (from fastkafka[docs,test]) (2.27.1)\n",
      "Requirement already satisfied: PyYAML>=5.3.1 in /usr/local/lib/python3.10/dist-packages (from fastkafka[docs,test]) (6.0)\n",
      "Collecting aiohttp>=3.8.4 (from fastkafka[docs,test])\n",
      "  Downloading aiohttp-3.8.4-cp310-cp310-manylinux_2_17_x86_64.manylinux2014_x86_64.whl (1.0 MB)\n",
      "\u001b[2K     \u001b[90m━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━\u001b[0m \u001b[32m1.0/1.0 MB\u001b[0m \u001b[31m51.2 MB/s\u001b[0m eta \u001b[36m0:00:00\u001b[0m\n",
      "\u001b[?25hRequirement already satisfied: attrs>=17.3.0 in /usr/local/lib/python3.10/dist-packages (from aiohttp>=3.8.4->fastkafka[docs,test]) (23.1.0)\n",
      "Requirement already satisfied: charset-normalizer<4.0,>=2.0 in /usr/local/lib/python3.10/dist-packages (from aiohttp>=3.8.4->fastkafka[docs,test]) (2.0.12)\n",
      "Collecting multidict<7.0,>=4.5 (from aiohttp>=3.8.4->fastkafka[docs,test])\n",
      "  Downloading multidict-6.0.4-cp310-cp310-manylinux_2_17_x86_64.manylinux2014_x86_64.whl (114 kB)\n",
      "\u001b[2K     \u001b[90m━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━\u001b[0m \u001b[32m114.5/114.5 kB\u001b[0m \u001b[31m4.6 MB/s\u001b[0m eta \u001b[36m0:00:00\u001b[0m\n",
      "\u001b[?25hCollecting async-timeout<5.0,>=4.0.0a3 (from aiohttp>=3.8.4->fastkafka[docs,test])\n",
      "  Downloading async_timeout-4.0.2-py3-none-any.whl (5.8 kB)\n",
      "Collecting yarl<2.0,>=1.0 (from aiohttp>=3.8.4->fastkafka[docs,test])\n",
      "  Downloading yarl-1.9.2-cp310-cp310-manylinux_2_17_x86_64.manylinux2014_x86_64.whl (268 kB)\n",
      "\u001b[2K     \u001b[90m━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━\u001b[0m \u001b[32m268.8/268.8 kB\u001b[0m \u001b[31m23.0 MB/s\u001b[0m eta \u001b[36m0:00:00\u001b[0m\n",
      "\u001b[?25hCollecting frozenlist>=1.1.1 (from aiohttp>=3.8.4->fastkafka[docs,test])\n",
      "  Downloading frozenlist-1.3.3-cp310-cp310-manylinux_2_5_x86_64.manylinux1_x86_64.manylinux_2_17_x86_64.manylinux2014_x86_64.whl (149 kB)\n",
      "\u001b[2K     \u001b[90m━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━\u001b[0m \u001b[32m149.6/149.6 kB\u001b[0m \u001b[31m15.5 MB/s\u001b[0m eta \u001b[36m0:00:00\u001b[0m\n",
      "\u001b[?25hCollecting aiosignal>=1.1.2 (from aiohttp>=3.8.4->fastkafka[docs,test])\n",
      "  Downloading aiosignal-1.3.1-py3-none-any.whl (7.6 kB)\n",
      "Collecting kafka-python>=2.0.2 (from aiokafka>=0.8.0->fastkafka[docs,test])\n",
      "  Downloading kafka_python-2.0.2-py2.py3-none-any.whl (246 kB)\n",
      "\u001b[2K     \u001b[90m━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━\u001b[0m \u001b[32m246.5/246.5 kB\u001b[0m \u001b[31m26.1 MB/s\u001b[0m eta \u001b[36m0:00:00\u001b[0m\n",
      "\u001b[?25hRequirement already satisfied: packaging in /usr/local/lib/python3.10/dist-packages (from aiokafka>=0.8.0->fastkafka[docs,test]) (23.1)\n",
      "Requirement already satisfied: idna>=2.8 in /usr/local/lib/python3.10/dist-packages (from anyio>=3.0->fastkafka[docs,test]) (3.4)\n",
      "Requirement already satisfied: sniffio>=1.1 in /usr/local/lib/python3.10/dist-packages (from anyio>=3.0->fastkafka[docs,test]) (1.3.0)\n",
      "Requirement already satisfied: ipykernel>=4.5.1 in /usr/local/lib/python3.10/dist-packages (from ipywidgets<=8.0.4,>=8.0->fastkafka[docs,test]) (5.5.6)\n",
      "Requirement already satisfied: ipython>=6.1.0 in /usr/local/lib/python3.10/dist-packages (from ipywidgets<=8.0.4,>=8.0->fastkafka[docs,test]) (7.34.0)\n",
      "Requirement already satisfied: traitlets>=4.3.1 in /usr/local/lib/python3.10/dist-packages (from ipywidgets<=8.0.4,>=8.0->fastkafka[docs,test]) (5.7.1)\n",
      "Collecting widgetsnbextension~=4.0 (from ipywidgets<=8.0.4,>=8.0->fastkafka[docs,test])\n",
      "  Downloading widgetsnbextension-4.0.7-py3-none-any.whl (2.1 MB)\n",
      "\u001b[2K     \u001b[90m━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━\u001b[0m \u001b[32m2.1/2.1 MB\u001b[0m \u001b[31m74.6 MB/s\u001b[0m eta \u001b[36m0:00:00\u001b[0m\n",
      "\u001b[?25hRequirement already satisfied: jupyterlab-widgets~=3.0 in /usr/local/lib/python3.10/dist-packages (from ipywidgets<=8.0.4,>=8.0->fastkafka[docs,test]) (3.0.7)\n",
      "Requirement already satisfied: typing-extensions>=4.2.0 in /usr/local/lib/python3.10/dist-packages (from pydantic>=1.9->fastkafka[docs,test]) (4.5.0)\n",
      "Requirement already satisfied: urllib3<1.27,>=1.21.1 in /usr/local/lib/python3.10/dist-packages (from requests>=2.20->fastkafka[docs,test]) (1.26.15)\n",
      "Requirement already satisfied: certifi>=2017.4.17 in /usr/local/lib/python3.10/dist-packages (from requests>=2.20->fastkafka[docs,test]) (2022.12.7)\n",
      "Requirement already satisfied: click<9.0.0,>=7.1.1 in /usr/local/lib/python3.10/dist-packages (from typer>=0.7.0->fastkafka[docs,test]) (8.1.3)\n",
      "Requirement already satisfied: ipython-genutils in /usr/local/lib/python3.10/dist-packages (from ipykernel>=4.5.1->ipywidgets<=8.0.4,>=8.0->fastkafka[docs,test]) (0.2.0)\n",
      "Requirement already satisfied: jupyter-client in /usr/local/lib/python3.10/dist-packages (from ipykernel>=4.5.1->ipywidgets<=8.0.4,>=8.0->fastkafka[docs,test]) (6.1.12)\n",
      "Requirement already satisfied: tornado>=4.2 in /usr/local/lib/python3.10/dist-packages (from ipykernel>=4.5.1->ipywidgets<=8.0.4,>=8.0->fastkafka[docs,test]) (6.3.1)\n",
      "Requirement already satisfied: setuptools>=18.5 in /usr/local/lib/python3.10/dist-packages (from ipython>=6.1.0->ipywidgets<=8.0.4,>=8.0->fastkafka[docs,test]) (67.7.2)\n",
      "Collecting jedi>=0.16 (from ipython>=6.1.0->ipywidgets<=8.0.4,>=8.0->fastkafka[docs,test])\n",
      "  Downloading jedi-0.18.2-py2.py3-none-any.whl (1.6 MB)\n",
      "\u001b[2K     \u001b[90m━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━\u001b[0m \u001b[32m1.6/1.6 MB\u001b[0m \u001b[31m74.1 MB/s\u001b[0m eta \u001b[36m0:00:00\u001b[0m\n",
      "\u001b[?25hRequirement already satisfied: decorator in /usr/local/lib/python3.10/dist-packages (from ipython>=6.1.0->ipywidgets<=8.0.4,>=8.0->fastkafka[docs,test]) (4.4.2)\n",
      "Requirement already satisfied: pickleshare in /usr/local/lib/python3.10/dist-packages (from ipython>=6.1.0->ipywidgets<=8.0.4,>=8.0->fastkafka[docs,test]) (0.7.5)\n",
      "Requirement already satisfied: prompt-toolkit!=3.0.0,!=3.0.1,<3.1.0,>=2.0.0 in /usr/local/lib/python3.10/dist-packages (from ipython>=6.1.0->ipywidgets<=8.0.4,>=8.0->fastkafka[docs,test]) (3.0.38)\n",
      "Requirement already satisfied: pygments in /usr/local/lib/python3.10/dist-packages (from ipython>=6.1.0->ipywidgets<=8.0.4,>=8.0->fastkafka[docs,test]) (2.14.0)\n",
      "Requirement already satisfied: backcall in /usr/local/lib/python3.10/dist-packages (from ipython>=6.1.0->ipywidgets<=8.0.4,>=8.0->fastkafka[docs,test]) (0.2.0)\n",
      "Requirement already satisfied: matplotlib-inline in /usr/local/lib/python3.10/dist-packages (from ipython>=6.1.0->ipywidgets<=8.0.4,>=8.0->fastkafka[docs,test]) (0.1.6)\n",
      "Requirement already satisfied: pexpect>4.3 in /usr/local/lib/python3.10/dist-packages (from ipython>=6.1.0->ipywidgets<=8.0.4,>=8.0->fastkafka[docs,test]) (4.8.0)\n",
      "Requirement already satisfied: parso<0.9.0,>=0.8.0 in /usr/local/lib/python3.10/dist-packages (from jedi>=0.16->ipython>=6.1.0->ipywidgets<=8.0.4,>=8.0->fastkafka[docs,test]) (0.8.3)\n",
      "Requirement already satisfied: ptyprocess>=0.5 in /usr/local/lib/python3.10/dist-packages (from pexpect>4.3->ipython>=6.1.0->ipywidgets<=8.0.4,>=8.0->fastkafka[docs,test]) (0.7.0)\n",
      "Requirement already satisfied: wcwidth in /usr/local/lib/python3.10/dist-packages (from prompt-toolkit!=3.0.0,!=3.0.1,<3.1.0,>=2.0.0->ipython>=6.1.0->ipywidgets<=8.0.4,>=8.0->fastkafka[docs,test]) (0.2.6)\n",
      "Requirement already satisfied: jupyter-core>=4.6.0 in /usr/local/lib/python3.10/dist-packages (from jupyter-client->ipykernel>=4.5.1->ipywidgets<=8.0.4,>=8.0->fastkafka[docs,test]) (5.3.0)\n",
      "Requirement already satisfied: pyzmq>=13 in /usr/local/lib/python3.10/dist-packages (from jupyter-client->ipykernel>=4.5.1->ipywidgets<=8.0.4,>=8.0->fastkafka[docs,test]) (23.2.1)\n",
      "Requirement already satisfied: python-dateutil>=2.1 in /usr/local/lib/python3.10/dist-packages (from jupyter-client->ipykernel>=4.5.1->ipywidgets<=8.0.4,>=8.0->fastkafka[docs,test]) (2.8.2)\n",
      "Requirement already satisfied: platformdirs>=2.5 in /usr/local/lib/python3.10/dist-packages (from jupyter-core>=4.6.0->jupyter-client->ipykernel>=4.5.1->ipywidgets<=8.0.4,>=8.0->fastkafka[docs,test]) (3.3.0)\n",
      "Requirement already satisfied: six>=1.5 in /usr/local/lib/python3.10/dist-packages (from python-dateutil>=2.1->jupyter-client->ipykernel>=4.5.1->ipywidgets<=8.0.4,>=8.0->fastkafka[docs,test]) (1.16.0)\n",
      "Building wheels for collected packages: install-jdk\n",
      "  Building wheel for install-jdk (setup.py) ... \u001b[?25l\u001b[?25hdone\n",
      "  Created wheel for install-jdk: filename=install_jdk-0.3.0-py3-none-any.whl size=3725 sha256=d5dad71fae09d32b4f9eeeb2f754a45ea49e7f948c48fc2184ccca9b5dd69dd1\n",
      "  Stored in directory: /root/.cache/pip/wheels/79/7a/47/9a4619174f7ca0f1068edb7a5412730a37365b6d183b0b3847\n",
      "Successfully built install-jdk\n",
      "Installing collected packages: kafka-python, widgetsnbextension, multidict, jedi, install-jdk, frozenlist, docstring-parser, async-timeout, yarl, asyncer, aiosignal, aiokafka, fastkafka, aiohttp, ipywidgets\n",
      "  Attempting uninstall: widgetsnbextension\n",
      "    Found existing installation: widgetsnbextension 3.6.4\n",
      "    Uninstalling widgetsnbextension-3.6.4:\n",
      "      Successfully uninstalled widgetsnbextension-3.6.4\n",
      "  Attempting uninstall: ipywidgets\n",
      "    Found existing installation: ipywidgets 7.7.1\n",
      "    Uninstalling ipywidgets-7.7.1:\n",
      "      Successfully uninstalled ipywidgets-7.7.1\n",
      "Successfully installed aiohttp-3.8.4 aiokafka-0.8.0 aiosignal-1.3.1 async-timeout-4.0.2 asyncer-0.0.2 docstring-parser-0.15 fastkafka-0.6.1 frozenlist-1.3.3 install-jdk-0.3.0 ipywidgets-8.0.4 jedi-0.18.2 kafka-python-2.0.2 multidict-6.0.4 widgetsnbextension-4.0.7 yarl-1.9.2\n"
     ]
    }
   ],
   "source": [
    "# | hide\n",
    "# | notest\n",
    "\n",
    "# Install FastKafka with testing and documentation dependencies\n",
    "\n",
    "! pip install fastkafka[test,docs]"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Tutorial\n",
    "\n",
    "You can start an interactive tutorial in Google Colab by clicking the button below:\n",
    "\n",
    "<a href=\"https://colab.research.google.com/github/airtai/fastkafka/blob/main/nbs/index.ipynb\" target=”_blank”>\n",
    "  <img src=\"https://colab.research.google.com/assets/colab-badge.svg\" alt=\"Open in Colab\" />\n",
    "</a>"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Writing server code\n",
    "\n",
    "To demonstrate FastKafka simplicity of using `@produces` and `@consumes` decorators, we will focus on a simple app.\n",
    "\n",
    "The app will consume JSON messages containing positive floats from one topic, log\n",
    "them, and then produce incremented values to another topic."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Messages\n",
    "\n",
    "FastKafka uses [Pydantic](https://docs.pydantic.dev/) to parse input JSON-encoded data into Python objects, making it easy to work with structured data in your Kafka-based applications. Pydantic's [`BaseModel`](https://docs.pydantic.dev/usage/models/) class allows you to define messages using a declarative syntax, making it easy to specify the fields and types of your messages.\n",
    "\n",
    "This example defines one `Data` mesage class. This Class will model the consumed and produced data in our app demo, it contains one `NonNegativeFloat` field `data` that will be logged and \"processed\" before being produced to another topic.\n",
    "\n",
    "These message class will be used to parse and validate incoming data in Kafka consumers and producers."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pydantic import BaseModel, Field, NonNegativeFloat\n",
    "\n",
    "\n",
    "class Data(BaseModel):\n",
    "    data: NonNegativeFloat = Field(\n",
    "        ..., example=0.5, description=\"Float data example\"\n",
    "    )"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Application\n",
    "\n",
    "This example shows how to initialize a FastKafka application.\n",
    "\n",
    "It starts by defining  a dictionary called `kafka_brokers`, which contains two entries: `\"localhost\"` and `\"production\"`, specifying local development and production Kafka brokers. Each entry specifies the URL, port, and other details of a Kafka broker. This dictionary is used for both generating the documentation and later to run the actual server against one of the given kafka broker.\n",
    "\n",
    "Next, an object of the `FastKafka` class is initialized with the minimum set of arguments:\n",
    "\n",
    "- `kafka_brokers`: a dictionary used for generation of documentation\n",
    "\n",
    "We will also import and create a logger so that we can log the incoming data in our consuming function."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from logging import getLogger\n",
    "from fastkafka import FastKafka\n",
    "\n",
    "logger = getLogger(\"Demo Kafka app\")\n",
    "\n",
    "kafka_brokers = {\n",
    "    \"localhost\": {\n",
    "        \"url\": \"localhost\",\n",
    "        \"description\": \"local development kafka broker\",\n",
    "        \"port\": 9092,\n",
    "    },\n",
    "    \"production\": {\n",
    "        \"url\": \"kafka.airt.ai\",\n",
    "        \"description\": \"production kafka broker\",\n",
    "        \"port\": 9092,\n",
    "        \"protocol\": \"kafka-secure\",\n",
    "        \"security\": {\"type\": \"plain\"},\n",
    "    },\n",
    "}\n",
    "\n",
    "kafka_app = FastKafka(\n",
    "    title=\"Demo Kafka app\",\n",
    "    kafka_brokers=kafka_brokers,\n",
    ")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Function decorators\n",
    "\n",
    "FastKafka provides convenient function decorators `@kafka_app.consumes` and `@kafka_app.produces` to allow you to delegate the actual process of\n",
    "\n",
    "- consuming and producing data to Kafka, and\n",
    "\n",
    "- decoding and encoding JSON messages\n",
    "\n",
    "from user defined functions to the framework. The FastKafka framework delegates these jobs to AIOKafka and Pydantic libraries.\n",
    "\n",
    "These decorators make it easy to specify the processing logic for your Kafka consumers and producers, allowing you to focus on the core business logic of your application without worrying about the underlying Kafka integration."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "This following example shows how to use the `@kafka_app.consumes` and `@kafka_app.produces` decorators in a FastKafka application:\n",
    "\n",
    "- The `@kafka_app.consumes` decorator is applied to the `on_input_data` function, which specifies that this function should be called whenever a message is received on the \"input_data\" Kafka topic. The `on_input_data` function takes a single argument which is expected to be an instance of the `Data` message class. Specifying the type of the single argument is instructing the Pydantic to use `Data.parse_raw()` on the consumed message before passing it to the user defined function `on_input_data`.\n",
    "\n",
    "- The `@produces` decorator is applied to the `to_output_data` function, which specifies that this function should produce a message to the \"output_data\" Kafka topic whenever it is called. The `to_output_data` function takes a single float argument `data`. It it increments the data returns it wrapped in a `Data` object. The framework will call the `Data.json().encode(\"utf-8\")` function on the returned value and produce it to the specified topic."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "@kafka_app.consumes(topic=\"input_data\", auto_offset_reset=\"latest\")\n",
    "async def on_input_data(msg: Data):\n",
    "    logger.info(f\"Got data: {msg.data}\")\n",
    "    await to_output_data(msg.data)\n",
    "\n",
    "\n",
    "@kafka_app.produces(topic=\"output_data\")\n",
    "async def to_output_data(data: float) -> Data:\n",
    "    processed_data = Data(data=data+1.0)\n",
    "    return processed_data"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Testing the service\n",
    "\n",
    "The service can be tested using the `Tester` instances which internally starts InMemory implementation of Kafka broker.\n",
    "\n",
    "The Tester will redirect your consumes and produces decorated functions to the InMemory Kafka broker so that you can quickly test your app without the need for a running Kafka broker and all its dependencies."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[INFO] fastkafka._testing.in_memory_broker: InMemoryBroker._start() called\n",
      "[INFO] fastkafka._testing.in_memory_broker: InMemoryBroker._patch_consumers_and_producers(): Patching consumers and producers!\n",
      "[INFO] fastkafka._testing.in_memory_broker: InMemoryBroker starting\n",
      "[INFO] fastkafka._application.app: _create_producer() : created producer using the config: '{'bootstrap_servers': 'localhost:9092'}'\n",
      "[INFO] fastkafka._testing.in_memory_broker: AIOKafkaProducer patched start() called()\n",
      "[INFO] fastkafka._application.app: _create_producer() : created producer using the config: '{'bootstrap_servers': 'localhost:9092'}'\n",
      "[INFO] fastkafka._testing.in_memory_broker: AIOKafkaProducer patched start() called()\n",
      "[INFO] fastkafka._components.aiokafka_consumer_loop: aiokafka_consumer_loop() starting...\n",
      "[INFO] fastkafka._components.aiokafka_consumer_loop: aiokafka_consumer_loop(): Consumer created using the following parameters: {'bootstrap_servers': 'localhost:9092', 'auto_offset_reset': 'latest', 'max_poll_records': 100}\n",
      "[INFO] fastkafka._testing.in_memory_broker: AIOKafkaConsumer patched start() called()\n",
      "[INFO] fastkafka._components.aiokafka_consumer_loop: aiokafka_consumer_loop(): Consumer started.\n",
      "[INFO] fastkafka._testing.in_memory_broker: AIOKafkaConsumer patched subscribe() called\n",
      "[INFO] fastkafka._testing.in_memory_broker: AIOKafkaConsumer.subscribe(), subscribing to: ['input_data']\n",
      "[INFO] fastkafka._components.aiokafka_consumer_loop: aiokafka_consumer_loop(): Consumer subscribed.\n",
      "[INFO] fastkafka._components.aiokafka_consumer_loop: aiokafka_consumer_loop() starting...\n",
      "[INFO] fastkafka._components.aiokafka_consumer_loop: aiokafka_consumer_loop(): Consumer created using the following parameters: {'bootstrap_servers': 'localhost:9092', 'auto_offset_reset': 'earliest', 'max_poll_records': 100}\n",
      "[INFO] fastkafka._testing.in_memory_broker: AIOKafkaConsumer patched start() called()\n",
      "[INFO] fastkafka._components.aiokafka_consumer_loop: aiokafka_consumer_loop(): Consumer started.\n",
      "[INFO] fastkafka._testing.in_memory_broker: AIOKafkaConsumer patched subscribe() called\n",
      "[INFO] fastkafka._testing.in_memory_broker: AIOKafkaConsumer.subscribe(), subscribing to: ['output_data']\n",
      "[INFO] fastkafka._components.aiokafka_consumer_loop: aiokafka_consumer_loop(): Consumer subscribed.\n",
      "[INFO] Demo Kafka app: Got data: 0.1\n",
      "[INFO] fastkafka._testing.in_memory_broker: AIOKafkaConsumer patched stop() called\n",
      "[INFO] fastkafka._components.aiokafka_consumer_loop: aiokafka_consumer_loop(): Consumer stopped.\n",
      "[INFO] fastkafka._components.aiokafka_consumer_loop: aiokafka_consumer_loop() finished.\n",
      "[INFO] fastkafka._testing.in_memory_broker: AIOKafkaProducer patched stop() called\n",
      "[INFO] fastkafka._testing.in_memory_broker: AIOKafkaConsumer patched stop() called\n",
      "[INFO] fastkafka._components.aiokafka_consumer_loop: aiokafka_consumer_loop(): Consumer stopped.\n",
      "[INFO] fastkafka._components.aiokafka_consumer_loop: aiokafka_consumer_loop() finished.\n",
      "[INFO] fastkafka._testing.in_memory_broker: AIOKafkaProducer patched stop() called\n",
      "[INFO] fastkafka._testing.in_memory_broker: InMemoryBroker._stop() called\n",
      "[INFO] fastkafka._testing.in_memory_broker: InMemoryBroker stopping\n"
     ]
    }
   ],
   "source": [
    "from fastkafka.testing import Tester\n",
    "\n",
    "msg = Data(\n",
    "    data=0.1,\n",
    ")\n",
    "\n",
    "# Start Tester app and create InMemory Kafka broker for testing\n",
    "async with Tester(kafka_app) as tester:\n",
    "    # Send Data message to input_data topic\n",
    "    await tester.to_input_data(msg)\n",
    "\n",
    "    # Assert that the kafka_app responded with incremented data in output_data topic\n",
    "    await tester.awaited_mocks.on_output_data.assert_awaited_with(\n",
    "        Data(data=1.1), timeout=2\n",
    "    )"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Recap\n",
    "\n",
    "We have created a simple FastKafka application.\n",
    "The app will consume the `Data` from the `input_data` topic, log it and produce the incremented data to `output_data` topic.\n",
    "\n",
    "To test the app we have:\n",
    "\n",
    "1. Created the app\n",
    "\n",
    "2. Started our Tester class which mirrors the developed app topics for testing purposes\n",
    "\n",
    "3. Sent Data message to `input_data` topic\n",
    "\n",
    "4. Asserted and checked that the developed service has reacted to Data message "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Running the service\n",
    "\n",
    "The service can be started using builtin faskafka run CLI command. Before we can do that, we will concatenate the code snippets from above and save them in a file `\"application.py\"`"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/markdown": [
       "\n",
       "```python\n",
       "# content of the \"application.py\" file\n",
       "\n",
       "from pydantic import BaseModel, Field, NonNegativeFloat\n",
       "\n",
       "from fastkafka import FastKafka\n",
       "from fastkafka._components.logger import get_logger\n",
       "\n",
       "logger = get_logger(__name__)\n",
       "\n",
       "class Data(BaseModel):\n",
       "    data: NonNegativeFloat = Field(\n",
       "        ..., example=0.5, description=\"Float data example\"\n",
       "    )\n",
       "\n",
       "kafka_brokers = {\n",
       "    \"localhost\": {\n",
       "        \"url\": \"localhost\",\n",
       "        \"description\": \"local development kafka broker\",\n",
       "        \"port\": 9092,\n",
       "    },\n",
       "    \"production\": {\n",
       "        \"url\": \"kafka.airt.ai\",\n",
       "        \"description\": \"production kafka broker\",\n",
       "        \"port\": 9092,\n",
       "        \"protocol\": \"kafka-secure\",\n",
       "        \"security\": {\"type\": \"plain\"},\n",
       "    },\n",
       "}\n",
       "\n",
       "kafka_app = FastKafka(\n",
       "    title=\"Demo Kafka app\",\n",
       "    kafka_brokers=kafka_brokers,\n",
       ")\n",
       "\n",
       "@kafka_app.consumes(topic=\"input_data\", auto_offset_reset=\"latest\")\n",
       "async def on_input_data(msg: Data):\n",
       "    logger.info(f\"Got data: {msg.data}\")\n",
       "    await to_output_data(msg.data)\n",
       "\n",
       "\n",
       "@kafka_app.produces(topic=\"output_data\")\n",
       "async def to_output_data(data: float) -> Data:\n",
       "    processed_data = Data(data=data+1.0)\n",
       "    return processed_data\n",
       "\n",
       "```\n"
      ],
      "text/plain": [
       "<IPython.core.display.Markdown object>"
      ]
     },
     "execution_count": null,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# | echo: false\n",
    "\n",
    "kafka_app_source = \"\"\"\n",
    "from pydantic import BaseModel, Field, NonNegativeFloat\n",
    "\n",
    "from fastkafka import FastKafka\n",
    "from fastkafka._components.logger import get_logger\n",
    "\n",
    "logger = get_logger(__name__)\n",
    "\n",
    "class Data(BaseModel):\n",
    "    data: NonNegativeFloat = Field(\n",
    "        ..., example=0.5, description=\"Float data example\"\n",
    "    )\n",
    "\n",
    "kafka_brokers = {\n",
    "    \"localhost\": {\n",
    "        \"url\": \"localhost\",\n",
    "        \"description\": \"local development kafka broker\",\n",
    "        \"port\": 9092,\n",
    "    },\n",
    "    \"production\": {\n",
    "        \"url\": \"kafka.airt.ai\",\n",
    "        \"description\": \"production kafka broker\",\n",
    "        \"port\": 9092,\n",
    "        \"protocol\": \"kafka-secure\",\n",
    "        \"security\": {\"type\": \"plain\"},\n",
    "    },\n",
    "}\n",
    "\n",
    "kafka_app = FastKafka(\n",
    "    title=\"Demo Kafka app\",\n",
    "    kafka_brokers=kafka_brokers,\n",
    ")\n",
    "\n",
    "@kafka_app.consumes(topic=\"input_data\", auto_offset_reset=\"latest\")\n",
    "async def on_input_data(msg: Data):\n",
    "    logger.info(f\"Got data: {msg.data}\")\n",
    "    await to_output_data(msg.data)\n",
    "\n",
    "\n",
    "@kafka_app.produces(topic=\"output_data\")\n",
    "async def to_output_data(data: float) -> Data:\n",
    "    processed_data = Data(data=data+1.0)\n",
    "    return processed_data\n",
    "\"\"\"\n",
    "\n",
    "with open(\"application.py\", \"w\") as source:\n",
    "    source.write(kafka_app_source)\n",
    "\n",
    "Markdown(\n",
    "    f\"\"\"\n",
    "```python\n",
    "# content of the \"application.py\" file\n",
    "{kafka_app_source}\n",
    "```\n",
    "\"\"\"\n",
    ")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "23-05-31 11:36:20.813 [INFO] fastkafka._components.test_dependencies: Installing Java...\n",
      "23-05-31 11:36:20.813 [INFO] fastkafka._components.test_dependencies:  - installing jdk...\n",
      "23-05-31 11:36:29.252 [INFO] fastkafka._components.test_dependencies:  - jdk path: /root/.jdk/jdk-11.0.19+7\n",
      "23-05-31 11:36:29.252 [INFO] fastkafka._components.test_dependencies: Java installed.\n",
      "23-05-31 11:36:29.356 [INFO] fastkafka._components.test_dependencies: Installing Kafka...\n",
      "832969it [00:05, 163016.44it/s]                \n",
      "23-05-31 11:36:35.571 [INFO] fastkafka._components.test_dependencies: Kafka installed in /root/.local/kafka_2.13-3.3.2.\n"
     ]
    }
   ],
   "source": [
    "# | notest\n",
    "# | hide\n",
    "\n",
    "# We need to start a local Apache Kafka broker so that our service can connect to it\n",
    "\n",
    "# To do that, we need JRE and Kafka toolkit installed.\n",
    "# We can do the installation by running \"fastkafka testing install_deps\"\n",
    "\n",
    "! fastkafka testing install_deps"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[INFO] fastkafka._testing.apache_kafka_broker: ApacheKafkaBroker.start(): entering...\n",
      "[WARNING] fastkafka._testing.apache_kafka_broker: ApacheKafkaBroker.start(): (<_UnixSelectorEventLoop running=True closed=False debug=False>) is already running!\n",
      "[WARNING] fastkafka._testing.apache_kafka_broker: ApacheKafkaBroker.start(): calling nest_asyncio.apply()\n",
      "[INFO] fastkafka._components.test_dependencies: Java is already installed.\n",
      "[INFO] fastkafka._components.test_dependencies: Kafka is installed.\n",
      "[INFO] fastkafka._components.test_dependencies: But not exported to PATH, exporting...\n",
      "[INFO] fastkafka._testing.apache_kafka_broker: Starting zookeeper...\n",
      "[INFO] fastkafka._testing.apache_kafka_broker: Starting kafka...\n",
      "[INFO] fastkafka._testing.apache_kafka_broker: Local Kafka broker up and running on 127.0.0.1:9092\n",
      "[INFO] fastkafka._testing.apache_kafka_broker: <class 'fastkafka.testing.ApacheKafkaBroker'>.start(): returning 127.0.0.1:9092\n",
      "[INFO] fastkafka._testing.apache_kafka_broker: ApacheKafkaBroker.start(): exited.\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "'127.0.0.1:9092'"
      ]
     },
     "execution_count": null,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# | notest\n",
    "# | hide\n",
    "\n",
    "# Start the local Apache Kafka broker\n",
    "\n",
    "from fastkafka.testing import ApacheKafkaBroker\n",
    "\n",
    "broker = ApacheKafkaBroker(apply_nest_asyncio=True)\n",
    "\n",
    "broker.start()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "To run the service, use the FastKafka CLI command and pass the module (in this case, the file where the app implementation is located) and the app simbol to the command.\n",
    "\n",
    "```sh\n",
    "fastkafka run --num-workers=1 --kafka-broker localhost application:kafka_app\n",
    "```\n",
    "\n",
    "After running the command, you should see the following output in your command line:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/markdown": [
       "\n",
       "**The cell blow will run your app indefinitely, you will need to stop it manually**\n"
      ],
      "text/plain": [
       "<IPython.core.display.Markdown object>"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "# | notest\n",
    "# | hide\n",
    "\n",
    "if in_colab:\n",
    "    display(Markdown(f\"\"\"\n",
    "**The cell blow will run your app indefinitely, you will need to stop it manually**\n",
    "\"\"\"))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[1504]: 23-05-31 11:36:45.874 [INFO] fastkafka._application.app: set_kafka_broker() : Setting bootstrap_servers value to 'localhost:9092'\n",
      "[1504]: 23-05-31 11:36:45.875 [INFO] fastkafka._application.app: _create_producer() : created producer using the config: '{'bootstrap_servers': 'localhost:9092'}'\n",
      "[1504]: 23-05-31 11:36:45.937 [INFO] fastkafka._components.aiokafka_consumer_loop: aiokafka_consumer_loop() starting...\n",
      "[1504]: 23-05-31 11:36:45.937 [INFO] fastkafka._components.aiokafka_consumer_loop: aiokafka_consumer_loop(): Consumer created using the following parameters: {'bootstrap_servers': 'localhost:9092', 'auto_offset_reset': 'latest', 'max_poll_records': 100}\n",
      "[1504]: 23-05-31 11:36:45.956 [INFO] fastkafka._components.aiokafka_consumer_loop: aiokafka_consumer_loop(): Consumer started.\n",
      "[1504]: 23-05-31 11:36:45.956 [INFO] aiokafka.consumer.subscription_state: Updating subscribed topics to: frozenset({'input_data'})\n",
      "[1504]: 23-05-31 11:36:45.956 [INFO] aiokafka.consumer.consumer: Subscribed to topic(s): {'input_data'}\n",
      "[1504]: 23-05-31 11:36:45.956 [INFO] fastkafka._components.aiokafka_consumer_loop: aiokafka_consumer_loop(): Consumer subscribed.\n",
      "[1506]: 23-05-31 11:36:45.993 [INFO] fastkafka._application.app: set_kafka_broker() : Setting bootstrap_servers value to 'localhost:9092'\n",
      "[1506]: 23-05-31 11:36:45.994 [INFO] fastkafka._application.app: _create_producer() : created producer using the config: '{'bootstrap_servers': 'localhost:9092'}'\n",
      "[1506]: 23-05-31 11:36:46.014 [INFO] fastkafka._components.aiokafka_consumer_loop: aiokafka_consumer_loop() starting...\n",
      "[1506]: 23-05-31 11:36:46.015 [INFO] fastkafka._components.aiokafka_consumer_loop: aiokafka_consumer_loop(): Consumer created using the following parameters: {'bootstrap_servers': 'localhost:9092', 'auto_offset_reset': 'latest', 'max_poll_records': 100}\n",
      "[1506]: 23-05-31 11:36:46.040 [INFO] fastkafka._components.aiokafka_consumer_loop: aiokafka_consumer_loop(): Consumer started.\n",
      "[1506]: 23-05-31 11:36:46.042 [INFO] aiokafka.consumer.subscription_state: Updating subscribed topics to: frozenset({'input_data'})\n",
      "[1506]: 23-05-31 11:36:46.043 [INFO] aiokafka.consumer.consumer: Subscribed to topic(s): {'input_data'}\n",
      "[1506]: 23-05-31 11:36:46.043 [INFO] fastkafka._components.aiokafka_consumer_loop: aiokafka_consumer_loop(): Consumer subscribed.\n",
      "[1506]: 23-05-31 11:36:46.068 [ERROR] aiokafka.cluster: Topic input_data not found in cluster metadata\n",
      "[1506]: 23-05-31 11:36:46.070 [INFO] aiokafka.consumer.group_coordinator: Metadata for topic has changed from {} to {'input_data': 0}. \n",
      "[1504]: 23-05-31 11:36:46.131 [WARNING] aiokafka.cluster: Topic input_data is not available during auto-create initialization\n",
      "[1504]: 23-05-31 11:36:46.132 [INFO] aiokafka.consumer.group_coordinator: Metadata for topic has changed from {} to {'input_data': 0}. \n",
      "[1506]: 23-05-31 11:37:00.237 [ERROR] aiokafka: Unable connect to node with id 0: [Errno 111] Connect call failed ('172.28.0.12', 9092)\n",
      "[1506]: 23-05-31 11:37:00.237 [ERROR] aiokafka: Unable to update metadata from [0]\n",
      "[1504]: 23-05-31 11:37:00.238 [ERROR] aiokafka: Unable connect to node with id 0: [Errno 111] Connect call failed ('172.28.0.12', 9092)\n",
      "[1504]: 23-05-31 11:37:00.238 [ERROR] aiokafka: Unable to update metadata from [0]\n",
      "[1506]: 23-05-31 11:37:00.294 [INFO] fastkafka._components.aiokafka_consumer_loop: aiokafka_consumer_loop(): Consumer stopped.\n",
      "[1506]: 23-05-31 11:37:00.294 [INFO] fastkafka._components.aiokafka_consumer_loop: aiokafka_consumer_loop() finished.\n",
      "Starting process cleanup, this may take a few seconds...\n",
      "23-05-31 11:37:00.345 [INFO] fastkafka._server: terminate_asyncio_process(): Terminating the process 1504...\n",
      "23-05-31 11:37:00.345 [INFO] fastkafka._server: terminate_asyncio_process(): Terminating the process 1506...\n",
      "[1504]: 23-05-31 11:37:00.347 [INFO] fastkafka._components.aiokafka_consumer_loop: aiokafka_consumer_loop(): Consumer stopped.\n",
      "[1504]: 23-05-31 11:37:00.347 [INFO] fastkafka._components.aiokafka_consumer_loop: aiokafka_consumer_loop() finished.\n",
      "23-05-31 11:37:00.607 [INFO] fastkafka._server: terminate_asyncio_process(): Process 1506 was already terminated.\n",
      "23-05-31 11:37:00.822 [INFO] fastkafka._server: terminate_asyncio_process(): Process 1504 was already terminated.\n"
     ]
    }
   ],
   "source": [
    "# | notest\n",
    "# | echo: false\n",
    "\n",
    "# This cell will run your app indefinitely, you need to stop it manually\n",
    "\n",
    "!fastkafka run --num-workers=2 --kafka-broker localhost application:kafka_app"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[INFO] fastkafka._testing.apache_kafka_broker: ApacheKafkaBroker.stop(): entering...\n",
      "[INFO] fastkafka._components._subprocess: terminate_asyncio_process(): Terminating the process 1068...\n",
      "[INFO] fastkafka._components._subprocess: terminate_asyncio_process(): Process 1068 was already terminated.\n",
      "[INFO] fastkafka._components._subprocess: terminate_asyncio_process(): Terminating the process 697...\n",
      "[INFO] fastkafka._components._subprocess: terminate_asyncio_process(): Process 697 was already terminated.\n",
      "[INFO] fastkafka._testing.apache_kafka_broker: ApacheKafkaBroker.stop(): exited.\n"
     ]
    }
   ],
   "source": [
    "# | notest\n",
    "# | hide\n",
    "\n",
    "# Stop the local broker to keep the runtime clean\n",
    "\n",
    "broker.stop()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Documentation\n",
    "\n",
    "The kafka app comes with builtin documentation generation using [AsyncApi HTML generator](https://www.asyncapi.com/tools/generator).\n",
    "\n",
    "AsyncApi requires Node.js to be installed and we provide the following convenience command line for it:\n",
    "\n",
    "```sh\n",
    "fastkafka docs install_deps\n",
    "```"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/markdown": [
       "\n",
       "# Update Node.js\n",
       "When running in Colab, we need to update Node.js first:\n",
       "    "
      ],
      "text/plain": [
       "<IPython.core.display.Markdown object>"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "# | notest\n",
    "# | hide\n",
    "\n",
    "if in_colab:\n",
    "    display(Markdown(\"\"\"\n",
    "# Update Node.js\n",
    "When running in Colab, we need to update Node.js first:\n",
    "    \"\"\"))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "\u001b[K\u001b[?25h/tools/node/bin/n -> /tools/node/lib/node_modules/n/bin/n\n",
      "+ n@9.1.0\n",
      "added 1 package from 2 contributors in 0.489s\n",
      "  \u001b[36minstalling\u001b[0m : \u001b[2mnode-v18.16.0\u001b[0m\n",
      "  \u001b[36m     mkdir\u001b[0m : \u001b[2m/usr/local/n/versions/node/18.16.0\u001b[0m\n",
      "  \u001b[36m     fetch\u001b[0m : \u001b[2mhttps://nodejs.org/dist/v18.16.0/node-v18.16.0-linux-x64.tar.xz\u001b[0m\n",
      "######################################################################## 100.0%\n",
      "\u001b[1A\u001b[2K  \u001b[36m   copying\u001b[0m : \u001b[2mnode/18.16.0\u001b[0m\n",
      "  \u001b[36m installed\u001b[0m : \u001b[2mv18.16.0 (with npm 9.5.1)\u001b[0m\n",
      "\n",
      "Note: the node command changed location and the old location may be remembered in your current shell.\n",
      "  \u001b[36m       old\u001b[0m : \u001b[2m/tools/node/bin/node\u001b[0m\n",
      "  \u001b[36m       new\u001b[0m : \u001b[2m/usr/local/bin/node\u001b[0m\n",
      "If \"node --version\" shows the old version then start a new shell, or reset the location hash with:\n",
      "hash -r  (for bash, zsh, ash, dash, and ksh)\n",
      "rehash   (for csh and tcsh)\n"
     ]
    }
   ],
   "source": [
    "# | notest\n",
    "# | hide\n",
    "\n",
    "if in_colab:\n",
    "    !npm install -g n\n",
    "    !n lts"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "23-05-31 11:38:24.128 [INFO] fastkafka._components.docs_dependencies: AsyncAPI generator installed\n"
     ]
    }
   ],
   "source": [
    "# | notest\n",
    "# | echo: false\n",
    "\n",
    "! fastkafka docs install_deps"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "To generate the documentation programatically you just need to call the following command:\n",
    "\n",
    "```sh\n",
    "fastkafka docs generate application:kafka_app\n",
    "```"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "23-05-31 11:38:25.113 [INFO] fastkafka._components.asyncapi: Old async specifications at '/content/asyncapi/spec/asyncapi.yml' does not exist.\n",
      "23-05-31 11:38:25.118 [INFO] fastkafka._components.asyncapi: New async specifications generated at: '/content/asyncapi/spec/asyncapi.yml'\n",
      "23-05-31 11:38:43.455 [INFO] fastkafka._components.asyncapi: Async docs generated at 'asyncapi/docs'\n",
      "23-05-31 11:38:43.455 [INFO] fastkafka._components.asyncapi: Output of '$ npx -y -p @asyncapi/generator ag asyncapi/spec/asyncapi.yml @asyncapi/html-template -o asyncapi/docs --force-write'\u001b[32m\n",
      "\n",
      "Done! ✨\u001b[0m\n",
      "\u001b[33mCheck out your shiny new generated files at \u001b[0m\u001b[35m/content/asyncapi/docs\u001b[0m\u001b[33m.\u001b[0m\n",
      "\n",
      "\n"
     ]
    }
   ],
   "source": [
    "# | notest\n",
    "# | echo: false\n",
    "\n",
    "! fastkafka docs generate application:kafka_app"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "This will generate the *asyncapi* folder in relative path where all your documentation will be saved. You can check out the content of it with:\n",
    "\n",
    "```sh\n",
    "ls -l asyncapi\n",
    "```"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "total 8\n",
      "drwxr-xr-x 4 root root 4096 May 31 11:38 docs\n",
      "drwxr-xr-x 2 root root 4096 May 31 11:38 spec\n"
     ]
    }
   ],
   "source": [
    "# | notest\n",
    "# | echo: false\n",
    "\n",
    "! ls -l asyncapi"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "In docs folder you will find the servable static html file of your documentation. This can also be served using our `fastkafka docs serve` CLI command (more on that in our guides).\n",
    "\n",
    "In spec folder you will find a asyncapi.yml file containing the async API specification of your application. "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We can locally preview the generated documentation by running the following command:\n",
    "\n",
    "```sh\n",
    "fastkafka docs serve application:kafka_app\n",
    "```"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/markdown": [
       "\n",
       "### Documentation link when runnining in Colab\n",
       "After running the cell below, when the cell outputs \"Serving documentation on http://127.0.0.1:8000\", you can acces your documentation with [this link](https://sd8jqq9m9u-496ff2e9c6d22116-8000-colab.googleusercontent.com/) \n",
       "\n",
       "**The cell below will serve docs indefinitely, you will need to stop it manually**\n"
      ],
      "text/plain": [
       "<IPython.core.display.Markdown object>"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "# | notest\n",
    "# | hide\n",
    "\n",
    "if in_colab:\n",
    "    port_proxy = google.colab.output.eval_js(f\"google.colab.kernel.proxyPort({8000})\")\n",
    "\n",
    "    display(Markdown(f\"\"\"\n",
    "### Documentation link when runnining in Colab\n",
    "After running the cell below, when the cell outputs \"Serving documentation on http://127.0.0.1:8000\", you can acces your documentation with [this link]({port_proxy}) \n",
    "\n",
    "**The cell below will serve docs indefinitely, you will need to stop it manually**\n",
    "\"\"\"))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "23-05-31 11:38:45.250 [INFO] fastkafka._components.asyncapi: New async specifications generated at: '/content/asyncapi/spec/asyncapi.yml'\n",
      "23-05-31 11:39:04.410 [INFO] fastkafka._components.asyncapi: Async docs generated at 'asyncapi/docs'\n",
      "23-05-31 11:39:04.411 [INFO] fastkafka._components.asyncapi: Output of '$ npx -y -p @asyncapi/generator ag asyncapi/spec/asyncapi.yml @asyncapi/html-template -o asyncapi/docs --force-write'\u001b[32m\n",
      "\n",
      "Done! ✨\u001b[0m\n",
      "\u001b[33mCheck out your shiny new generated files at \u001b[0m\u001b[35m/content/asyncapi/docs\u001b[0m\u001b[33m.\u001b[0m\n",
      "\n",
      "\n",
      "Serving documentation on http://127.0.0.1:8000\u001b[0m\n",
      "127.0.0.1 - - [31/May/2023 11:39:14] \"GET / HTTP/1.1\" 200 -\n",
      "127.0.0.1 - - [31/May/2023 11:39:14] \"GET /css/global.min.css HTTP/1.1\" 200 -\n",
      "127.0.0.1 - - [31/May/2023 11:39:14] \"GET /js/asyncapi-ui.min.js HTTP/1.1\" 200 -\n",
      "127.0.0.1 - - [31/May/2023 11:39:14] \"GET /css/asyncapi.min.css HTTP/1.1\" 200 -\n",
      "Interupting serving of documentation and cleaning up...\u001b[0m\n"
     ]
    }
   ],
   "source": [
    "# | notest\n",
    "# | echo: false\n",
    "\n",
    "# This cell will serve docs indefinitely, you need to stop it manually\n",
    "\n",
    "!fastkafka docs serve application:kafka_app"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "From the parameters passed to the application constructor, we get the documentation bellow:\n",
    "```python\n",
    "from fastkafka import FastKafka\n",
    "\n",
    "kafka_brokers = {\n",
    "    \"localhost\": {\n",
    "        \"url\": \"localhost\",\n",
    "        \"description\": \"local development kafka broker\",\n",
    "        \"port\": 9092,\n",
    "    },\n",
    "    \"production\": {\n",
    "        \"url\": \"kafka.airt.ai\",\n",
    "        \"description\": \"production kafka broker\",\n",
    "        \"port\": 9092,\n",
    "        \"protocol\": \"kafka-secure\",\n",
    "        \"security\": {\"type\": \"plain\"},\n",
    "    },\n",
    "}\n",
    "\n",
    "kafka_app = FastKafka(\n",
    "    title=\"Demo Kafka app\",\n",
    "    kafka_brokers=kafka_brokers,\n",
    ")\n",
    "```"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "![Kafka_servers](https://raw.githubusercontent.com/airtai/fastkafka/main/nbs/images/screenshot-kafka-servers.png)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The following documentation snippet are for the consumer as specified in the code above:"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "\n",
    "![Kafka_consumer](https://raw.githubusercontent.com/airtai/fastkafka/main/nbs/images/screenshot-kafka-consumer.png)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The following documentation snippet are for the producer as specified in the code above:"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "\n",
    "![Kafka_producer](https://raw.githubusercontent.com/airtai/fastkafka/main/nbs/images/screenshot-kafka-producer.png)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Finally, all messages as defined as subclasses of *BaseModel* are documented as well: "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "\n",
    "![Kafka_![Kafka_servers](https://raw.githubusercontent.com/airtai/fastkafka/main/nbs/images/screenshot-kafka-messages.png)](https://raw.githubusercontent.com/airtai/fastkafka/main/nbs/images/screenshot-kafka-messages.png)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## License\n",
    "\n",
    "FastKafka is licensed under the Apache License 2.0\n",
    "\n",
    "A permissive license whose main conditions require preservation of copyright and license notices. Contributors provide an express grant of patent rights. Licensed works, modifications, and larger works may be distributed under different terms and without source code.\n",
    "\n",
    "The full text of the license can be found [here](https://raw.githubusercontent.com/airtai/fastkafka/main/LICENSE)."
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "python3",
   "language": "python",
   "name": "python3"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 1
}
